package com.bw.gmall.realtime.dws.app;

import com.bw.gmall.realtime.common.base.BaseSqlApp;
import com.bw.gmall.realtime.common.constant.Constant;
import com.bw.gmall.realtime.common.util.SQLUtil;
import com.bw.gmall.realtime.dws.function.KwSplit;
import javafx.scene.control.Tab;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableConfig;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Author sw
 * @CreateTime 2024-11-19
 * 12.1 流量域搜索关键词粒度页面浏览各窗口汇总表
 */
public class DwsTrafficSourceKeywordPageViewWindow extends BaseSqlApp {
    public static void main(String[] args) {
        new DwsTrafficSourceKeywordPageViewWindow().start(Constant.DWS_TRAFFIC_SOURCE_KEYWORD_PAGE_VIEW_WINDOW,4,10021);
    }

    @Override
    public void handle(StreamExecutionEnvironment env, StreamTableEnvironment tableEnv, String groupId) {
        TableConfig config = tableEnv.getConfig();
        config.getConfiguration().setBoolean("table.exec.emit.early-fire.enabled",true);
        config.getConfiguration().setString("table.exec.emit.early-fire.delay","5s");

        //1、读取日志表搜索的数据
        getPageLog(tableEnv);
        //2、获取搜索商品的数据
        getWord(tableEnv);
//        tableEnv.sqlQuery("select * from word").execute().print();
        tableEnv.createTemporaryFunction("SplitFunction", KwSplit.class);
        //3、炸开函数
        getExplodeWord(tableEnv);
        //4、开窗聚合
//        tableEnv.sqlQuery("select * from explodeWord").execute().print();
        Table table = getSumWord(tableEnv);
//        tableEnv.sqlQuery("select * from word").execute().print();
//        table.execute().print();
        //5、写出Doris
        extracted(tableEnv);
        table.insertInto(Constant.DWS_TRAFFIC_SOURCE_KEYWORD_PAGE_VIEW_WINDOW).execute();
    }

    private static void extracted(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql("CREATE TABLE dws_traffic_source_keyword_page_view_window (\n" +
                " stt  STRING,\n" +
                " edt STRING,\n" +
                " cur_date STRING,\n" +
                " keyword  STRING,\n" +
                " keyword_count BIGINT\n" +
                ")" + SQLUtil.getDorisSinkSQL(Constant.DWS_TRAFFIC_SOURCE_KEYWORD_PAGE_VIEW_WINDOW));
    }

    private static Table getSumWord(StreamTableEnvironment tableEnv) {
//        return tableEnv.sqlQuery(" SELECT\n" +
//                " date_format(window_start,'yyyy-MM-dd HH:mm:ss') window_start,\n" +
//                " date_format(window_end,'yyyy-MM-dd HH:mm:ss') window_end,\n" +
//                " date_format(now(),'yyyyMMdd') cur_date,\n" +
//                " word keyword,\n" +
//                " count(*) keyword_count\n" +
//                " FROM TABLE(\n" +
//                " TUMBLE(TABLE explodeWord, DESCRIPTOR(et), INTERVAL '5' SECOND))\n" +
//                " GROUP BY window_start, window_end,word");


        return tableEnv.sqlQuery("SELECT\n" +
                "  date_format(TUMBLE_START(et, interval '5' second),'yyyy-MM-dd HH:mm:ss') AS stt,\n" +
                "  date_format(TUMBLE_END(et, interval '5' second),'yyyy-MM-dd HH:mm:ss') AS edt,\n" +
                "  date_format(now() , 'yyyyMMdd') cur_date, " +
                "  word keyword,\n" +
                "  count(*) keyword_count\n" +
                "FROM explodeWord\n" +
                "GROUP BY\n" +
                "  TUMBLE(et, interval '5' second ),\n" +
                "  word");
    }

    private static void getExplodeWord(StreamTableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(" select\n" +
                " item, \n" +
                " word, \n" +
                " et\n" +
                " from word, LATERAL TABLE(SplitFunction(item))");
        tableEnv.createTemporaryView("explodeWord",table);
    }

    private static void getWord(StreamTableEnvironment tableEnv) {
        Table table = tableEnv.sqlQuery(" select\n" +
                " `page`['item'] item,\n" +
                " et\n" +
                " from page_log\n" +
                " where `page`['last_page_id'] = 'search'\n" +
                " and `page`['item_type'] = 'keyword'");
        tableEnv.createTemporaryView("word",table);
    }

    private static void getPageLog(StreamTableEnvironment tableEnv) {
        tableEnv.executeSql(" create table page_log(\n" +
                " page map<String,String>,\n" +
                " common map<String,String>,\n" +
                " ts bigint,\n" +
                " et as to_timestamp_ltz(ts,3),\n" +
                " WATERMARK FOR et AS et - INTERVAL '5' SECOND\n" +
                " )" + SQLUtil.getKafkaSourceSQL(Constant.TOPIC_DWD_TRAFFIC_PAGE,Constant.TOPIC_DWD_TRAFFIC_PAGE));
    }
}
